package com.song.elasticsearch.config.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;

import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.CompletableFuture;

//@Component
public class KafkaConsumerThread implements ApplicationRunner {
    @Autowired
    KafkaConsumer<String, String> kafkaConsumer;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        warningHandle();
    }

    public void warningHandle(){
        kafkaConsumer.subscribe((Collection<String>) Arrays.asList("kafka_ztyyk_hjrk_cswsb"));
        CompletableFuture.runAsync(() -> {
            while (true){
                ConsumerRecords<String, String> records = kafkaConsumer.poll(1000);
                for (ConsumerRecord<String, String> record : records){
                    System.out.println(String.format("topic: %s, message: %s", record.topic(), record.value()));
                }
            }
        });
    }
}
